﻿#include <core/network/network_node.h>
#include <core/task/task_interface.h>
#include <thirdparty/fist-log.h>
#include <util/fist_rand.h>

#include <cppcoro/sync_wait.hpp>
#include <cppcoro/single_consumer_event.hpp>
#include <utility>

#include "rpc_manager.h"

#include <ranges>

#include "rpc_conf.h"
#include "rpc_helper.h"


CRpcManager::CRpcManager()
    : m_last_send_alive(0)
      , m_last_co_event_check(0)
      , m_cfg_parser(new CRpcConfParser)
      , m_task_manager(nullptr) { }

CRpcManager::~ CRpcManager()
{
    delete m_cfg_parser;
}

bool CRpcManager::Init()
{
    if (!m_cfg_parser->Load("../config/server/core/rpc.xml")) {
        return false;
    }
    m_task_manager = CMngPtr<ITaskManager>(this).ptr();
    m_network_manager = CMngPtr<INetworkManager>(this).ptr();
    return true;
}


void CRpcManager::Update()
{
    m_cfg_parser->Update();
    CHECK_TIMEOUT(m_last_send_alive, conf()->keep_server_alive, KeepAlive);
    CHECK_TIMEOUT(m_last_co_event_check, conf()->co_event_timeout, ReleaseCoEventWhenTimeout);
}

void CRpcManager::Destroy() { }

void CRpcManager::ProcMessage(netlink_t link, const std::string& msg)
{
    RemoteCallMessage remoteMsg;
    remoteMsg.ParseFromString(msg);
    switch (remoteMsg.type()) {
        case RPC_MSG_TYPE_REMOTE_CALL:
            HandleRemoteCall(remoteMsg, link);
            break;
        case RPC_MSG_TYPE_REMOTE_CALL_RETURN:
            HandleRemoteCallReturn(remoteMsg, link);
            break;
        case RPC_MSG_TYPE_KEEP_ALIVE:
            HandleKeepAlive(remoteMsg, link);
            break;
        default:
            break;
    }
}

void CRpcManager::RemoteCall(RemoteCallMessage& msg)
{
    const auto& transmit = this->peers()->find(BOX_TYPE_TRANSMIT);
    if (transmit == this->peers()->end()) {
        LOG_ERROR("not find any transmit peer!");
        return;
    }
    protobuf_service_token* source = msg.source().New();
    source->set_id(service_id());
    source->set_type(service_type());
    msg.set_allocated_source(source);
    auto& peers = transmit->second;
    if (peers.size() == 1) {
        m_network_manager->SendOneMessage(peers[0].first, msg.SerializeAsString());
    } else {
        auto& target = peers[fist::simple_rand(static_cast<uint32_t>(peers.size()))];
        m_network_manager->SendOneMessage(target.first, msg.SerializeAsString());
    }
}

cppcoro::task<> CRpcManager::RemoteCoCall(RemoteCallMessage& msg, std::string& returnData)
{
    auto& record = m_co_events.emplace();
    record.release = fist::get_now_ms_time() + conf()->co_event_timeout;
    msg.set_co((uint64_t)(&record));
    //LOG_DEBUG("{}", (uint64_t)(&record));
    RemoteCall(msg);
    co_await record.lock;
    if (record.returned) {
        returnData.swap(record.return_data);
    }
    co_return;
}

void CRpcManager::HandleRemoteCallReturn(RemoteCallMessage& msg, const netlink_t& link)
{
    const auto& target = msg.target();

    if (msg.co() == 0) {
        LOG_ERROR("not co message! target[{}, {}]", target.type(), target.id());
        return;
    }
    //LOG_DEBUG("1 {} {} {}", msg.co(), target.type(), target.id());
    if (target.type() == service_type()) {
        // 目标服务器
        if (target.id() == BOX_TYPE_INVALID) {
            LOG_ERROR("invalid message! from_svc:{}", target.type());
            return;
        }
        if (target.id() != service_id()) {
            LOG_WARN("message ignore! target[type:{}, id:{}], self[type:{}, id:{}]", target.type(), target.id(),
                     service_type(), service_id());
            return;
        }
        const std::string& ret = msg.params().Get(0);
        uint64_t co = msg.co();
        auto task = [this, ret, co]()-> cppcoro::task<> {
            co_await HandleCoCallReturn(ret, co);
        };
        m_task_manager->PushNetworkTask(std::move(task));
    } else {
        // 转发
        int realId = target.id();
        netlink_t send = INVALID_LINK;
        if (target.id() != BOX_TYPE_INVALID) {
            send = TargetLink(target.type(), target.id());
        } else {
            send = RandomLink(target.type(), link, realId);
        }
        if (link == INVALID_LINK) {
            return;
        }
        //LOG_DEBUG("2 {} {} {} {}", msg.co(), target.type(), target.id(), realId);
        int ret = m_network_manager->SendOneMessage(send, SerializeAsString(msg));
        if (ret != NETWORK_OK) {
            LOG_ERROR("send msg error! send_svc:{} id:{}", target.type(), target.id());
        }
    }
}

netlink_t CRpcManager::RandomLink(const int& type, const netlink_t& except, int& realId)
{
    m_random_link_temp.clear();
    for (auto& peer : m_remotes | std::views::values) {
        if (peer.link == except) {
            continue;
        }
        if (auto node = static_cast<CNetworkNode*>(peer.link); !node->valid()) {
            continue;
        }
        m_random_link_temp.emplace_back(peer.link, peer.id);
    }
    if (m_random_link_temp.empty()) {
        LOG_ERROR("all server is invalid! type:{}", type);
        return INVALID_LINK;
    }
    if (m_random_link_temp.size() == 1) {
        realId = m_random_link_temp[0].second;
        return m_random_link_temp[0].first;
    }
    auto& rand = m_random_link_temp[fist::simple_rand(static_cast<uint32_t>(m_random_link_temp.size()))];
    realId = rand.second;
    return rand.first;
}

netlink_t CRpcManager::TargetLink(const int& type, const int& id)
{
    const auto& target = m_remotes.find({type, id});
    if (target == m_remotes.end()) {
        LOG_ERROR("not find target remote! {}, {}", type, id);
        return INVALID_LINK;
    }
    return target->second.link;
}

cppcoro::task<> CRpcManager::HandleCoCallReturn(const std::string& ret, uint64_t co)
{
    if (m_co_events.empty()) {
        LOG_ERROR("message co not found!");
        co_return;
    }
    auto& back = m_co_events.back();
    if (back.release < fist::get_now_ms_time() - conf()->delay_del_co_event) {
        LOG_ERROR("message co not found!");
        co_return;
    }
    RpcRemoteCoCallRecord* record = (RpcRemoteCoCallRecord*)(co);
    record->return_data.assign(ret);
    record->returned = true;
    record->lock.set();
}

const RpcConf* CRpcManager::conf()
{
    return m_cfg_parser->conf();
}

void CRpcManager::HandleRemoteCall(RemoteCallMessage& msg, const netlink_t& link)
{
    auto& target = msg.target();
    if (target.type() != service_type()) {
        // 转发
        int realId = target.id();
        netlink_t send = INVALID_LINK;
        if (target.id() != BOX_TYPE_INVALID) {
            send = TargetLink(target.type(), target.id());
        } else {
            send = RandomLink(target.type(), link, realId);
        }
        //LOG_DEBUG("{} {} {}", target.type(), target.id(), realId);
        if (link == INVALID_LINK) {
            return;
        }
        protobuf_service_token* p = msg.mutable_target();
        p->set_type(target.type());
        p->set_id(realId);
        int ret = m_network_manager->SendOneMessage(send, SerializeAsString(msg));
        if (ret != NETWORK_OK) {
            LOG_ERROR("send msg error! svc_type:{} id:{}", p->type(), p->id());
        }
        return;
    }
    // 目标服务器
    if (target.id() != service_id()) {
        LOG_ERROR("message ignore! target[type:{}, id:{}], self[type:{}, id:{}]", target.type(), target.id(),
                  service_type(), service_id());
        return;
    }
    //LOG_DEBUG("1 {} {} {}", msg.co(), target.type(), target.id());
    if (msg.co() == 0) {
        auto p = QueryManager(msg.call_manager());
        if (p == nullptr) {
            LOG_ERROR("not find manager %llu", msg.call_manager());
            return;
        }
        auto rpc = static_cast<IRmiManager*>(p);
        rpc->__handle_remote_call__(msg);
    } else {
        m_task_manager->PushTask([&, this]()-> cppcoro::task<> {
            co_await HandleRemoteCoCall(msg, link);
        });
    }
}

cppcoro::task<> CRpcManager::HandleRemoteCoCall(RemoteCallMessage& msg, const netlink_t& link)
{
    auto p = QueryManager(msg.call_manager());
    if (p == nullptr) {
        LOG_ERROR("not find manager %llu", msg.call_manager());
        co_return;
    }
    auto rpc = static_cast<IRmiManager*>(p);
    returnDataCache.clear();
    co_await rpc->__handle_remote_co_call__(msg, returnDataCache);
    //LOG_DEBUG("1 {} {} {}", msg.co(), msg.source().type(), msg.source().id());
    // 将执行结果返回服务器
    RemoteCallMessage return_wrapper;
    return_wrapper.set_allocated_source(new protobuf_service_token(msg.target()));
    return_wrapper.set_allocated_target(new protobuf_service_token(msg.source()));
    return_wrapper.set_type(RPC_MSG_TYPE_REMOTE_CALL_RETURN);
    return_wrapper.set_call_manager(msg.call_manager());
    return_wrapper.set_call_func(msg.call_func());
    return_wrapper.set_co(msg.co());
    return_wrapper.add_params(returnDataCache);
    RemoteCall(return_wrapper);
    co_return;
}


void CRpcManager::HandleKeepAlive(RemoteCallMessage& msg, const netlink_t& link)
{
    const RpcMsgKeepLive keep_live = ParseFromString<RpcMsgKeepLive>(msg.params().Get(0));
    LOG_DEBUG("type:{}, id:{}", keep_live.from_svc(), keep_live.svc_id());
    if (auto f = m_remotes.find({keep_live.from_svc(), keep_live.svc_id()}); f == m_remotes.end()) {
        m_remotes.insert({
            service_token_t(keep_live.from_svc(), keep_live.svc_id()),
            RpcRemotePeerInfo(keep_live.from_svc(), keep_live.svc_id(), link, fist::get_now_ms_time())
        });
    } else {
        f->second.link = link;
        f->second.last_ack = fist::get_now_ms_time();
    }
}


void CRpcManager::KeepAlive()
{
    if (this->peers()->empty()) {
        return;
    }

    RpcMsgKeepLive keep_live;
    keep_live.set_from_svc(service_type());
    keep_live.set_svc_id(service_id());
    RemoteCallMessage call_wrapper;
    call_wrapper.set_type(RPC_MSG_TYPE_KEEP_ALIVE);
    call_wrapper.add_params(SerializeAsString(keep_live));

    for (auto peer : *(this->peers())) {
        for (auto link : peer.second) {
            m_network_manager->SendOneMessage(link.first, call_wrapper.SerializeAsString());
        }
    }
}

void CRpcManager::ReleaseCoEventWhenTimeout()
{
    fist::ms_t now = fist::get_now_ms_time() + conf()->delay_del_co_event;
    while (!m_co_events.empty()) {
        auto& top = m_co_events.front();
        if (top.release < now || top.returned) {
            top.lock.set();
            m_co_events.pop();
            continue;
        }
        return;
    }
}
